package rabbitmq

import (
	"context"
	"fmt"
	"github.com/wagslane/go-rabbitmq"
	"log"
	"time"

	"gitee.com/xwengf/go-event/v2/event"
)

var (
	_ event.Sender   = (*rabbitSender)(nil)
	_ event.Receiver = (*rabbitReceiver)(nil)
	_ event.Event    = (*Message)(nil)
)

type Message struct {
	key   string
	value []byte
}

func (m *Message) Key() string {
	return m.key
}

func (m *Message) Value() []byte {
	return m.value
}

func NewMessage(key string, value []byte) event.Event {
	return &Message{
		key:   key,
		value: value,
	}
}

// ====================== Sender ======================

type rabbitSender struct {
	publisher  *rabbitmq.Publisher
	exchange   string
	routingKey string
}

func (s *rabbitSender) Send(ctx context.Context, topic string, message event.Event) error {
	headers := rabbitmq.Table{
		"key": message.Key(),
	}
	return s.publisher.PublishWithContext(
		ctx,
		message.Value(),
		[]string{topic},
		rabbitmq.WithPublishOptionsExchange(s.exchange),
		rabbitmq.WithPublishOptionsPersistentDelivery,
		rabbitmq.WithPublishOptionsHeaders(headers),
	)
}

func (s *rabbitSender) Close() error {
	s.publisher.Close()
	return nil
}

func NewRabbitSender(amqpURL, exchange, routingKey string) (event.Sender, error) {
	conn, err := rabbitmq.NewConn(
		amqpURL,
		rabbitmq.WithConnectionOptionsReconnectInterval(2*time.Second),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create RabbitMQ connection: %w", err)
	}

	pub, err := rabbitmq.NewPublisher(
		conn,
		rabbitmq.WithPublisherOptionsExchangeName(exchange),
		rabbitmq.WithPublisherOptionsExchangeDeclare, // 自动声明交换机
		rabbitmq.WithPublisherOptionsExchangeDurable,
		rabbitmq.WithPublisherOptionsExchangeKind("direct"),
		rabbitmq.WithPublisherOptionsLogging,
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create publisher: %w", err)
	}

	return &rabbitSender{
		publisher:  pub,
		exchange:   exchange,
		routingKey: routingKey,
	}, nil
}

// ====================== Receiver ======================

type rabbitReceiver struct {
	consumer *rabbitmq.Consumer
	queue    string
	close    chan struct{}
}

func (r *rabbitReceiver) Receive(ctx context.Context, handler event.Handler) error {
	defer fmt.Println("[rabbitmq] receiver stopped!")

	defer r.consumer.Close()

	err := r.consumer.Run(func(d rabbitmq.Delivery) rabbitmq.Action {
		// rabbitmq.Ack, rabbitmq.NackDiscard, rabbitmq.NackRequeue
		key := ""
		if val, ok := d.Headers["key"]; ok {
			if s, ok := val.(string); ok {
				key = s
			}
		}
		err := handler(ctx, key, NewMessage(key, d.Body))
		if err != nil {
			// 消息处理失败，决定是否将其放到死信队列或重试
			fmt.Printf("‼️ 处理消息失败2: %v", err)
			return rabbitmq.NackDiscard
		}
		return rabbitmq.Ack
	})
	if err != nil {
		log.Fatal(err)
	}
	return nil
}

func (r *rabbitReceiver) Close() error {
	fmt.Println("[rabbitmq] receiver closing")
	r.consumer.Close()
	return nil
}

func NewRabbitReceiver(amqpURL, exchange, queue, routingKey string) (event.Receiver, error) {
	conn, err := rabbitmq.NewConn(
		amqpURL,
		rabbitmq.WithConnectionOptionsReconnectInterval(2*time.Second),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create RabbitMQ connection: %w", err)
	}

	consumer, err := rabbitmq.NewConsumer(
		conn,
		queue,
		rabbitmq.WithConsumerOptionsExchangeName(exchange),
		rabbitmq.WithConsumerOptionsExchangeDeclare,
		rabbitmq.WithConsumerOptionsExchangeDurable,
		rabbitmq.WithConsumerOptionsExchangeKind("direct"),
		rabbitmq.WithConsumerOptionsQueueDurable,
		rabbitmq.WithConsumerOptionsRoutingKey(routingKey),
		rabbitmq.WithConsumerOptionsConsumerAutoAck(false),
		rabbitmq.WithConsumerOptionsLogging,
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create consumer: %w", err)
	}

	return &rabbitReceiver{
		consumer: consumer,
		queue:    queue,
		close:    make(chan struct{}),
	}, nil
}
